package com.atguigu.flink.timeAndwindow;

import com.atguigu.flink.pojo.OrderDetailEvent;
import com.atguigu.flink.pojo.OrderEvent;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * Created by 黄凯 on 2023/6/18 0018 21:00
 *
 * @author 黄凯
 * 永远相信美好的事情总会发生.
 * <p>
 * 间隔联结 IntervalJoin
 * *
 * *   以一条流为主，设置一个时间间隔(以当前数据的时间 ，设置一个上界和下界) ,
 * 另外一条流中相同key的数据如果落到了间隔的范围内，就可以join成功。
 */
public class Flink12_IntervalJoin {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //orderEvent
        //order-1,1000
        SingleOutputStreamOperator<OrderEvent> orderDs = env.socketTextStream("127.0.0.1", 8888)
                .map(
                        line -> {
                            String[] fields = line.split(",");
                            return new OrderEvent(fields[0].trim(), Long.valueOf(fields[1].trim()));
                        }

                ).assignTimestampsAndWatermarks(
                        WatermarkStrategy.<OrderEvent>forBoundedOutOfOrderness(Duration.ZERO)
                                .withTimestampAssigner(
                                        (event, ts) -> event.getTs()
                                )
                );

        orderDs.print("orderDs-input");

        //OrderDetailEvent
        //detail-1,order-1,Apple,3000
        SingleOutputStreamOperator<OrderDetailEvent> detailDs = env.socketTextStream("127.0.0.1", 9999)
                .map(
                        line -> {
                            String[] fields = line.split(",");
                            return new OrderDetailEvent(fields[0].trim(), fields[1].trim(), fields[2].trim(), Long.valueOf(fields[3].trim()));
                        }

                ).assignTimestampsAndWatermarks(
                        WatermarkStrategy.<OrderDetailEvent>forBoundedOutOfOrderness(Duration.ZERO)
                                .withTimestampAssigner(
                                        (event, ts) -> event.getTs()
                                )
                );

        detailDs.print("detailDs-input");

        // order流 与 detail流的join
        orderDs.keyBy(
                        OrderEvent::getOrderId
                ).intervalJoin(
                        detailDs.keyBy(
                                OrderDetailEvent::getOrderId
                        )
                ).between(
                        Time.seconds(-2), Time.seconds(2)
                )
//                .upperBoundExclusive()
//                .lowerBoundExclusive()
                .process(
                        new ProcessJoinFunction<OrderEvent, OrderDetailEvent, String>() {

                            /**
                             * 处理join成功的数据
                             */
                            @Override
                            public void processElement(OrderEvent left, OrderDetailEvent right,
                                                       ProcessJoinFunction<OrderEvent, OrderDetailEvent, String>.Context ctx, Collector<String> out) throws Exception {

                                out.collect(left + " -- " + right);

                            }
                        }
                ).print();

        env.execute();

    }

}
